package com.example.rabbitmq.starter.service;

import com.example.rabbitmq.starter.dao.MQConfigRepository;
import com.example.rabbitmq.starter.entity.MQConfigInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import lombok.RequiredArgsConstructor;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;

/**
 * MQ配置服务类
 */
@Service
@RequiredArgsConstructor
public class MQConfigService {

    private static final Logger log = LoggerFactory.getLogger(MQConfigService.class);

    private final MQConfigRepository mqConfigRepository;

    private final RabbitAdmin rabbitAdmin;

    /**
     * 初始化所有已启用的配置
     */
    @PostConstruct
    public void init() {
        List<MQConfigInfo> enabledConfigs = mqConfigRepository.findByEnabledTrue();
        enabledConfigs.forEach(this::applyConfig);
        log.info("MQ配置初始化完成，共加载 {} 个配置", enabledConfigs.size());
    }

    /**
     * 创建MQ配置
     *
     * @param configInfo 配置信息
     * @return 创建后的配置信息
     */
    public MQConfigInfo createConfig(MQConfigInfo configInfo) {
        // 生成配置ID
        if (configInfo.getId() == null || configInfo.getId().isEmpty()) {
            configInfo.setId(UUID.randomUUID().toString());
        }

        // 验证交换机类型一致性
        validateExchangeTypeConsistency(configInfo, null);

        // 设置创建时间和更新时间
        Date now = new Date();
        configInfo.setCreateTime(now);
        configInfo.setUpdateTime(now);

        // 保存配置
        MQConfigInfo saved = mqConfigRepository.save(configInfo);

        // 如果配置已启用，则应用配置
        if (saved.isEnabled()) {
            applyConfig(saved);
        }

        return saved;
    }

    /**
     * 更新MQ配置
     *
     * @param configInfo 配置信息
     * @return 更新后的配置信息
     */
    public MQConfigInfo updateConfig(MQConfigInfo configInfo) {
        // 查询原配置
        Optional<MQConfigInfo> optionalOriginal = mqConfigRepository.findById(configInfo.getId());
        if (!optionalOriginal.isPresent()) {
            throw new IllegalArgumentException("配置不存在：" + configInfo.getId());
        }

        MQConfigInfo original = optionalOriginal.get();

        // 验证交换机类型一致性
        validateExchangeTypeConsistency(configInfo, configInfo.getId());

        // 设置更新时间
        configInfo.setCreateTime(original.getCreateTime());
        configInfo.setUpdateTime(new Date());

        // 保存配置
        MQConfigInfo saved = mqConfigRepository.save(configInfo);

        // 如果原配置已启用，则删除原配置
        if (original.isEnabled()) {
            removeConfig(original);
        }

        // 如果新配置已启用，则应用新配置
        if (saved.isEnabled()) {
            applyConfig(saved);
        }

        return saved;
    }

    /**
     * 启用MQ配置
     *
     * @param id 配置ID
     * @return 启用后的配置信息
     */
    public MQConfigInfo enableConfig(String id) {
        Optional<MQConfigInfo> optionalConfig = mqConfigRepository.findById(id);
        if (!optionalConfig.isPresent()) {
            throw new IllegalArgumentException("配置不存在：" + id);
        }

        MQConfigInfo configInfo = optionalConfig.get();
        if (!configInfo.isEnabled()) {
            configInfo.setEnabled(true);
            configInfo.setUpdateTime(new Date());
            mqConfigRepository.save(configInfo);

            // 应用配置
            applyConfig(configInfo);
        }

        return configInfo;
    }

    /**
     * 禁用MQ配置
     *
     * @param id 配置ID
     * @return 禁用后的配置信息
     */
    public MQConfigInfo disableConfig(String id) {
        Optional<MQConfigInfo> optionalConfig = mqConfigRepository.findById(id);
        if (!optionalConfig.isPresent()) {
            throw new IllegalArgumentException("配置不存在：" + id);
        }

        MQConfigInfo configInfo = optionalConfig.get();
        if (configInfo.isEnabled()) {
            configInfo.setEnabled(false);
            configInfo.setUpdateTime(new Date());
            mqConfigRepository.save(configInfo);

            // 移除配置
            removeConfig(configInfo);
        }

        return configInfo;
    }

    /**
     * 查询所有MQ配置
     *
     * @return 配置信息列表
     */
    public List<MQConfigInfo> getAllConfigs() {
        return mqConfigRepository.findAll();
    }

    /**
     * 根据ID查询MQ配置
     *
     * @param id 配置ID
     * @return 配置信息
     */
    public MQConfigInfo getConfigById(String id) {
        return mqConfigRepository.findById(id).orElse(null);
    }

    /**
     * 根据业务名称查询MQ配置
     *
     * @param businessName 业务名称
     * @return 配置信息
     */
    public MQConfigInfo getConfigByBusinessName(String businessName) {
        return mqConfigRepository.findByBusinessName(businessName);
    }

    /**
     * 根据消息类型查询MQ配置
     *
     * @param messageType 消息类型
     * @return 配置信息
     */
    public MQConfigInfo getConfigByMessageType(String messageType) {
        return mqConfigRepository.findByMessageType(messageType);
    }

    /**
     * 应用MQ配置
     *
     * @param configInfo 配置信息
     */
    private void applyConfig(MQConfigInfo configInfo) {
        try {
            // 创建交换机
            Exchange exchange = createExchange(configInfo);
            rabbitAdmin.declareExchange(exchange);

            // 创建队列
            org.springframework.amqp.core.Queue queue = createQueueFromConfig(configInfo);
            rabbitAdmin.declareQueue(queue);

            // 创建绑定关系
            Binding binding = createBinding(configInfo, exchange, queue);
            rabbitAdmin.declareBinding(binding);

            log.info("MQ配置应用成功：businessName={}, exchange={}, queue={}, routingKey={}",
                    configInfo.getBusinessName(), configInfo.getExchangeName(),
                    configInfo.getQueueName(), configInfo.getRoutingKey());
        }
        catch (Exception e) {
            log.error("MQ配置应用失败：businessName={}, error={}",
                    configInfo.getBusinessName(), e.getMessage(), e);

            // 设置配置为禁用状态
            configInfo.setEnabled(false);
            configInfo.setUpdateTime(new Date());
            mqConfigRepository.save(configInfo);
        }
    }

    /**
     * 移除MQ配置
     *
     * @param configInfo 配置信息
     */
    private void removeConfig(MQConfigInfo configInfo) {
        try {
            // 移除绑定关系
            rabbitAdmin.removeBinding(createBinding(configInfo,
                    createExchange(configInfo), createQueueFromConfig(configInfo)));

            // 移除队列
            rabbitAdmin.deleteQueue(configInfo.getQueueName());

            // 移除交换机
            rabbitAdmin.deleteExchange(configInfo.getExchangeName());

            log.info("MQ配置移除成功：businessName={}, exchange={}, queue={}",
                    configInfo.getBusinessName(), configInfo.getExchangeName(), configInfo.getQueueName());
        }
        catch (Exception e) {
            log.error("MQ配置移除失败：businessName={}, error={}",
                    configInfo.getBusinessName(), e.getMessage(), e);
        }
    }

    /**
     * 创建交换机
     *
     * @param configInfo 配置信息
     * @return 交换机对象
     */
    private Exchange createExchange(MQConfigInfo configInfo) {
        String type = configInfo.getExchangeType();
        String name = configInfo.getExchangeName();
        boolean durable = configInfo.isExchangeDurable();
        boolean autoDelete = false;
        Map<String, Object> arguments = configInfo.getExchangeArguments();

        switch (type) {
            case "direct":
                return new DirectExchange(name, durable, autoDelete, arguments);
            case "topic":
                return new TopicExchange(name, durable, autoDelete, arguments);
            case "fanout":
                return new FanoutExchange(name, durable, autoDelete, arguments);
            case "headers":
                return new HeadersExchange(name, durable, autoDelete, arguments);
            default:
                throw new IllegalArgumentException("不支持的交换机类型：" + type);
        }
    }

    /**
     * 创建队列
     *
     * @param configInfo 配置信息
     * @return 队列对象
     */
    private org.springframework.amqp.core.Queue createQueueFromConfig(MQConfigInfo configInfo) {
        Map<String, Object> arguments = new HashMap<>();

        // 复制原始参数
        if (configInfo.getQueueArguments() != null && !configInfo.getQueueArguments().isEmpty()) {
            arguments.putAll(configInfo.getQueueArguments());
        }

        return new org.springframework.amqp.core.Queue(
                configInfo.getQueueName(),
                configInfo.isQueueDurable(),
                false,  // exclusive
                configInfo.isQueueAutoDelete(),
                arguments
        );
    }

    /**
     * 创建绑定关系
     *
     * @param configInfo 配置信息
     * @param exchange   交换机对象
     * @param queue      队列对象
     * @return 绑定对象
     */
    private Binding createBinding(MQConfigInfo configInfo, Exchange exchange, org.springframework.amqp.core.Queue queue) {
        String type = configInfo.getExchangeType();

        switch (type) {
            case "direct":
                return BindingBuilder.bind(queue)
                        .to((DirectExchange) exchange)
                        .with(configInfo.getRoutingKey());
            case "topic":
                return BindingBuilder.bind(queue)
                        .to((TopicExchange) exchange)
                        .with(configInfo.getRoutingKey());
            case "fanout":
                return BindingBuilder.bind(queue)
                        .to((FanoutExchange) exchange);
            case "headers":
                return BindingBuilder.bind(queue)
                        .to((HeadersExchange) exchange)
                        .where("key").matches(configInfo.getRoutingKey());
            default:
                throw new IllegalArgumentException("不支持的交换机类型：" + type);
        }
    }

    /**
     * 分页查询MQ配置
     *
     * @param pageable 分页参数
     * @return 分页结果
     */
    public Page<MQConfigInfo> findAll(Pageable pageable) {
        return mqConfigRepository.findAll(pageable);
    }

    /**
     * 根据消息类型分页查询MQ配置
     *
     * @param messageType 消息类型
     * @param pageable    分页参数
     * @return 分页结果
     */
    public Page<MQConfigInfo> findByMessageType(String messageType, Pageable pageable) {
        return mqConfigRepository.findByMessageType(messageType, pageable);
    }

    /**
     * 根据消息类型分页查询MQ配置
     *
     * @param messageType 消息类型
     * @return 分页结果
     */
    public List<MQConfigInfo> findByMessageType(String messageType) {
        return mqConfigRepository.queryAllByMessageType(messageType);
    }

    /**
     * 根据启用状态分页查询MQ配置
     *
     * @param enabled  启用状态
     * @param pageable 分页参数
     * @return 分页结果
     */
    public Page<MQConfigInfo> findByEnabled(Boolean enabled, Pageable pageable) {
        return mqConfigRepository.findByEnabled(enabled, pageable);
    }

    /**
     * 根据消息类型和启用状态分页查询MQ配置
     *
     * @param messageType 消息类型
     * @param enabled     启用状态
     * @param pageable    分页参数
     * @return 分页结果
     */
    public Page<MQConfigInfo> findByMessageTypeAndEnabled(String messageType, Boolean enabled, Pageable pageable) {
        return mqConfigRepository.findByMessageTypeAndEnabled(messageType, enabled, pageable);
    }

    /**
     * 获取队列统计信息
     *
     * @return 队列统计信息列表
     */
    public List<Map<String, Object>> getQueueStats() {
        List<Map<String, Object>> result = new ArrayList<>();
        List<MQConfigInfo> configs = mqConfigRepository.findByEnabledTrue();

        for (MQConfigInfo config : configs) {
            try {
                String queueName = config.getQueueName();
                Map<String, Object> stats = new HashMap<>();

                // 获取队列属性
                Properties queueProperties = rabbitAdmin.getQueueProperties(queueName);
                if (queueProperties != null) {
                    // 基本信息
                    stats.put("queueName", queueName);
                    stats.put("businessName", config.getBusinessName());
                    stats.put("exchangeName", config.getExchangeName());
                    stats.put("routingKey", config.getRoutingKey());

                    // 消息数量统计
                    Number messageCount = (Number) queueProperties.get("QUEUE_MESSAGE_COUNT");
                    Number consumerCount = (Number) queueProperties.get("QUEUE_CONSUMER_COUNT");
                    Number messagesReady = (Number) queueProperties.get("QUEUE_MESSAGES_READY");
                    Number messagesUnacknowledged = (Number) queueProperties.get("QUEUE_MESSAGES_UNACKNOWLEDGED");

                    stats.put("messageCount", messageCount != null ? messageCount.intValue() : 0);
                    stats.put("consumerCount", consumerCount != null ? consumerCount.intValue() : 0);
                    stats.put("messagesReady", messagesReady != null ? messagesReady.intValue() : 0);
                    stats.put("messagesUnacknowledged", messagesUnacknowledged != null ? messagesUnacknowledged.intValue() : 0);

                    // 计算消息送达率和确认率
                    int totalMessages = messageCount != null ? messageCount.intValue() : 0;
                    int unackMessages = messagesUnacknowledged != null ? messagesUnacknowledged.intValue() : 0;
                    int readyMessages = messagesReady != null ? messagesReady.intValue() : 0;

                    // 如果有消息，计算送达率和确认率
                    int deliveryRate = 100;
                    int ackRate = 100;

                    if (totalMessages > 0) {
                        int processedMessages = totalMessages - readyMessages;
                        deliveryRate = (int) ((double) processedMessages / totalMessages * 100);

                        if (processedMessages > 0) {
                            int ackedMessages = processedMessages - unackMessages;
                            ackRate = (int) ((double) ackedMessages / processedMessages * 100);
                        }
                    }

                    stats.put("deliveryRate", deliveryRate);
                    stats.put("ackRate", ackRate);

                    result.add(stats);
                }
            }
            catch (Exception e) {
                log.error("获取队列[{}]统计信息失败: {}", config.getQueueName(), e.getMessage());
                // 继续处理下一个队列
            }
        }

        return result;
    }

    /**
     * 删除MQ配置（带死信队列保护）
     *
     * @param id 配置ID
     */
    public void deleteConfig(String id) {
        Optional<MQConfigInfo> optionalConfig = mqConfigRepository.findById(id);
        if (!optionalConfig.isPresent()) {
            throw new IllegalArgumentException("配置不存在：" + id);
        }

        MQConfigInfo configInfo = optionalConfig.get();

        // 如果配置已启用，则先移除配置
        if (configInfo.isEnabled()) {
            removeConfig(configInfo);
        }

        // 删除配置
        mqConfigRepository.deleteById(id);
        log.info("成功删除MQ配置[{}]", configInfo.getBusinessName());
    }

    /**
     * 验证交换机类型一致性
     * 确保同一个交换机名称只能有一种交换机类型
     *
     * @param configInfo 待验证的配置信息
     * @param excludeId  排除的配置ID（用于更新时排除自身）
     * @throws IllegalArgumentException 如果交换机类型不一致
     */
    private void validateExchangeTypeConsistency(MQConfigInfo configInfo, String excludeId) {
        if (configInfo.getExchangeName() == null || configInfo.getExchangeName().trim().isEmpty()) {
            throw new IllegalArgumentException("交换机名称不能为空");
        }

        if (configInfo.getExchangeType() == null || configInfo.getExchangeType().trim().isEmpty()) {
            throw new IllegalArgumentException("交换机类型不能为空");
        }

        // 查询同名交换机的所有配置
        List<MQConfigInfo> existingConfigs = mqConfigRepository.findByExchangeName(configInfo.getExchangeName());

        // 过滤掉当前配置（用于更新场景）
        if (excludeId != null) {
            existingConfigs = existingConfigs.stream()
                    .filter(config -> !excludeId.equals(config.getId()))
                    .collect(java.util.stream.Collectors.toList());
        }

        // 检查是否存在不同类型的交换机配置
        for (MQConfigInfo existingConfig : existingConfigs) {
            if (!configInfo.getExchangeType().equals(existingConfig.getExchangeType())) {
                throw new IllegalArgumentException(
                        String.format("交换机 '%s' 已存在类型为 '%s' 的配置，不能创建类型为 '%s' 的配置。同一个交换机只能有一种类型。",
                                configInfo.getExchangeName(),
                                existingConfig.getExchangeType(),
                                configInfo.getExchangeType())
                );

            }
        }

        log.debug("交换机类型一致性验证通过：交换机={}, 类型={}",
                configInfo.getExchangeName(), configInfo.getExchangeType());
    }

    public MQConfigInfo findById(String id) {
        return mqConfigRepository.findById(id).orElseThrow(() -> new IllegalArgumentException("配置不存在：" + id));
    }
}